package com.study.flink.dataset

import java.util

import org.apache.flink.api.common.functions.RichMapFunction
import org.apache.flink.api.scala.ExecutionEnvironment
import org.apache.flink.configuration.Configuration

/**
  * 广播变量，保存在每个节点的内存，不能太大
  *
  * @author stephen
  * @create 2019-05-26 23:44
  * @since 1.0.0
  */
object FlinkBroadcastVariablesDemo {

  def main(args: Array[String]): Unit = {
    // 1、执行环境
    val env = ExecutionEnvironment.getExecutionEnvironment

    // 2、指定数据源
    import org.apache.flink.api.scala._
    // step 1:定义用于广播变量的数据
    val toBroadcast = env.fromElements("a", "b")
    val dataStream = env.fromElements("hadoop", "spark", "storm", "flink")

    // 3、对数据进行操作
    val resultStream = dataStream.map(new RichMapFunction[String, String] {

      var broadcastSet: util.Collection[String] = null

      override def open(parameters: Configuration): Unit = {
        // step 3: 得到广播变量
        broadcastSet = getRuntimeContext.getBroadcastVariable("broadcastSetName")
      }

      override def map(in: String): String = {
        // 使用广播变量
        // todo
        in
      }
    })
      // step 2: 广播变量
      .withBroadcastSet(toBroadcast, "broadcastSetName")

    // 4、指定数据输出位置
    resultStream.print()

  }
}
